package org.apache.flink.kubernetes;

import static org.apache.flink.util.Preconditions.checkNotNull;

import java.util.Optional;

import org.apache.flink.client.cluster.ClusterClient;
import org.apache.flink.client.cluster.ClusterClientProvider;
import org.apache.flink.client.cluster.rest.RestClusterClient;
import org.apache.flink.client.deployment.ClusterDeploymentException;
import org.apache.flink.client.deployment.ClusterDescriptor;
import org.apache.flink.client.deployment.ClusterRetrieveException;
import org.apache.flink.client.deployment.ClusterSpecification;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.ConsumerManagerOptions;
import org.apache.flink.configuration.HighAvailabilityOptions;
import org.apache.flink.configuration.ProviderManagerOptions;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.kubernetes.configuration.KubernetesConfigOptions;
import org.apache.flink.kubernetes.configuration.KubernetesConfigOptionsInternal;
import org.apache.flink.kubernetes.entrypoint.KubernetesClusterEntrypoint;
import org.apache.flink.kubernetes.kubeclient.ComponentPod;
import org.apache.flink.kubernetes.kubeclient.ERKubeClient;
import org.apache.flink.kubernetes.kubeclient.Endpoint;
import org.apache.flink.kubernetes.kubeclient.KubernetesConsumerManagerSpecification;
import org.apache.flink.kubernetes.kubeclient.factory.KubernetesResourceConsumerFactory;
import org.apache.flink.kubernetes.kubeclient.parameters.KubernetesMasterComponentParameters;
import org.apache.flink.kubernetes.utils.Constants;
import org.apache.flink.kubernetes.utils.KubernetesUtils;
import org.apache.flink.runtime.consumermanager.HighAvailabilityMode;
import org.apache.flink.runtime.entrypoint.ClusterEntrypoint;
import org.apache.flink.runtime.highavailability.HighAvailabilityServicesUtils;
import org.apache.flink.runtime.highavailability.nonha.standalone.StandaloneClientHAServices;
import org.apache.flink.runtime.rpc.AddressResolution;
import org.apache.flink.util.ERException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/** Kubernetes specific {@link ClusterDescriptor} implementation. */
public class KubernetesClusterDescriptor implements ClusterDescriptor<String> {

    private static final Logger LOG = LoggerFactory.getLogger(KubernetesClusterDescriptor.class);

    private static final String CLUSTER_DESCRIPTION = "Kubernetes cluster";

    private final Configuration erConfig;

    private final ERKubeClient client;

    private final String clusterId;

    public KubernetesClusterDescriptor(Configuration erConfig, ERKubeClient client) {
        this.erConfig = erConfig;
        this.client = client;
        this.clusterId =
                checkNotNull(
                        erConfig.getString(KubernetesConfigOptions.CLUSTER_ID),
                        "ClusterId must be specified!");
    }

    @Override
    public String getClusterDescription() {
        return CLUSTER_DESCRIPTION;
    }

    private ClusterClientProvider<String> createClusterClientProvider(String clusterId) {
        return () -> {
            final Configuration configuration = new Configuration(erConfig);

            final Optional<Endpoint> restEndpoint = client.getRestEndpoint(clusterId);

            if (restEndpoint.isPresent()) {
                configuration.setString(RestOptions.ADDRESS, restEndpoint.get().getAddress());
                configuration.setInteger(RestOptions.PORT, restEndpoint.get().getPort());
            } else {
                throw new RuntimeException(
                        new ClusterRetrieveException(
                                "Could not get the rest endpoint of " + clusterId));
            }

            try {
                // client will always use Kubernetes service to contact with master component. So we
                // have a pre-configured web monitor address. Using StandaloneClientHAServices to
                // create RestClusterClient is reasonable.
                return new RestClusterClient<>(
                        configuration,
                        clusterId,
                        (effectiveConfiguration, fatalErrorHandler) ->
                                new StandaloneClientHAServices(
                                        getWebMonitorAddress(effectiveConfiguration)));
            } catch (Exception e) {
                throw new RuntimeException(
                        new ClusterRetrieveException("Could not create the RestClusterClient.", e));
            }
        };
    }

    private String getWebMonitorAddress(Configuration configuration) throws Exception {
        AddressResolution resolution = AddressResolution.TRY_ADDRESS_RESOLUTION;
        final KubernetesConfigOptions.ServiceExposedType serviceType =
                configuration.get(KubernetesConfigOptions.REST_SERVICE_EXPOSED_TYPE);
        if (serviceType.isClusterIP()) {
            resolution = AddressResolution.NO_ADDRESS_RESOLUTION;
            LOG.warn(
                    "Please note that Flink client operations(e.g. cancel, list, stop,"
                            + " savepoint, etc.) won't work from outside the Kubernetes cluster"
                            + " since '{}' has been set to {}.",
                    KubernetesConfigOptions.REST_SERVICE_EXPOSED_TYPE.key(),
                    serviceType);
        }
        return HighAvailabilityServicesUtils.getWebMonitorAddress(configuration, resolution);
    }

    @Override
    public ClusterClientProvider<String> retrieve(String clusterId) {
        final ClusterClientProvider<String> clusterClientProvider =
                createClusterClientProvider(clusterId);

        try (ClusterClient<String> clusterClient = clusterClientProvider.getClusterClient()) {
            LOG.info(
                    "Retrieve flink cluster {} successfully", clusterId);
        }
        return clusterClientProvider;
    }

    @Override
    public ClusterClientProvider<String> deployCluster(
            ClusterSpecification clusterSpecification) throws ClusterDeploymentException {
        final ClusterClientProvider<String> clusterClientProvider =
                deployClusterInternal(
                        KubernetesClusterEntrypoint.class.getName(),
                        clusterSpecification,
                        false);

        try (ClusterClient<String> clusterClient = clusterClientProvider.getClusterClient()) {
            LOG.info(
                    "Create flink session cluster {} successfully.",
                    clusterId);
        }
        return clusterClientProvider;
    }

    private ClusterClientProvider<String> deployClusterInternal(
            String entryPoint, ClusterSpecification clusterSpecification, boolean detached)
            throws ClusterDeploymentException {
        final ClusterEntrypoint.ExecutionMode executionMode =
                detached
                        ? ClusterEntrypoint.ExecutionMode.DETACHED
                        : ClusterEntrypoint.ExecutionMode.NORMAL;
        erConfig.setString(
                ClusterEntrypoint.INTERNAL_CLUSTER_EXECUTION_MODE, executionMode.toString());
        // 入口点
        erConfig.setString(KubernetesConfigOptionsInternal.ENTRY_POINT_CLASS, entryPoint);
        // Rpc, rest, taskManagerRpc ports need to be exposed, so update them to fixed values.
        KubernetesUtils.checkAndUpdatePortConfigOption(
                erConfig, ProviderManagerOptions.RPC_PORT, Constants.TASK_MANAGER_RPC_PORT);
        KubernetesUtils.checkAndUpdatePortConfigOption(
                erConfig, RestOptions.BIND_PORT, Constants.REST_PORT);
        // 高可用
        if (HighAvailabilityMode.isHighAvailabilityModeActivated(erConfig)) {
            erConfig.setString(HighAvailabilityOptions.HA_CLUSTER_ID, clusterId);
            KubernetesUtils.checkAndUpdatePortConfigOption(
                    erConfig,
                    HighAvailabilityOptions.HA_JOB_MANAGER_PORT_RANGE,
                    erConfig.get(ConsumerManagerOptions.PORT));
        }

        try {
        	// 构建作业管理器参数
            final KubernetesMasterComponentParameters kubernetesJobManagerParameters =
                    new KubernetesMasterComponentParameters(erConfig, clusterSpecification);
            // 获取作业管理器通用配置(模板)
            final ComponentPod podTemplate =
                    kubernetesJobManagerParameters
                            .getPodTemplateFilePath()
                            .map(file ->KubernetesUtils.loadPodFromTemplateFile(
                                                    client, file, Constants.MAIN_CONTAINER_NAME))
                            .orElse(new ComponentPod.Builder().build());
            // 构建作业管理器部署规格 
            final KubernetesConsumerManagerSpecification kubernetesJobManagerSpec =
                    KubernetesResourceConsumerFactory.buildKubernetesResourceConsumerSpecification(
                            podTemplate, kubernetesJobManagerParameters);
            // 部署作业管理器
            client.createConsumerManagerComponent(kubernetesJobManagerSpec);
            // 返回cluster client
            return createClusterClientProvider(clusterId);
        } catch (Exception e) {
            try {
                LOG.warn(
                        "Failed to create the Kubernetes cluster \"{}\", try to clean up the residual resources.",
                        clusterId);
                client.stopAndCleanupCluster(clusterId);
            } catch (Exception e1) {
                LOG.info(
                        "Failed to stop and clean up the Kubernetes cluster \"{}\".",
                        clusterId,
                        e1);
            }
            throw new ClusterDeploymentException(
                    "Could not create Kubernetes cluster \"" + clusterId + "\".", e);
        }
    }

    @Override
    public void killCluster(String clusterId) throws ERException {
        try {
            client.stopAndCleanupCluster(clusterId);
        } catch (Exception e) {
            throw new ERException("Could not kill Kubernetes cluster " + clusterId);
        }
    }

    @Override
    public void close() {
        try {
            client.close();
        } catch (Exception e) {
            LOG.error("failed to close client, exception {}", e.toString());
        }
    }
}
